package com.maven.rabbitmq.confirm;

import com.maven.rabbitmq.constant.Constant;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class PublishConfirm {
     public final static long len =1000;

    public static Connection creatConnection() throws Exception {
        //1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("8.135.54.49");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("rabbitmq");

        return  connectionFactory.newConnection();
    }


    public static void main(String[] args) throws Exception {
        // 单独确认
//        publishMessagesIndividually();
        // 批量确认
//        publishMessagesInbatch();
        // 异步确认
        publishMessagesAsynchronously();
    }




    public static void publishMessagesIndividually() throws Exception{
        try(Connection connection= PublishConfirm.creatConnection()){
            Channel channel=connection.createChannel();

            // 开启信道确认模式
            channel.confirmSelect();
            channel.queueDeclare(Constant.INDIVIDUALLY_QUEUE,true,false,false,null);

            long start=System.currentTimeMillis();
            for(int i = 0; i< len; i++){
                String msg = " hello I'm individually confirm "+ i;
                channel.basicPublish("",Constant.INDIVIDUALLY_QUEUE,null,msg.getBytes());

                // 当我们开启单独确认之后，如果超时未返回就会抛出 Timeout 异常
                // 如果 消息被确认就会正常返回
                channel.waitForConfirms(5000);
            }

            long end=System.currentTimeMillis();
            System.out.println("batch 花费： "+ (end-start)+"ms");
        }
    }


    public static void publishMessagesInbatch() throws Exception {
        try (Connection connection = creatConnection()) {
            Channel channel = connection.createChannel();

            // 开启信道确认模式
            channel.confirmSelect();
            channel.queueDeclare(Constant.INBATCH_QUEUE, true, false, false, null);

            long start = System.currentTimeMillis();
            int batchSize=100;
            int currentSize=0;
            for (int i = 0; i < len; i++) {
                String msg = " hello I'm individually confirm " + i;
                channel.basicPublish("", Constant.INBATCH_QUEUE, null, msg.getBytes());
                currentSize++;

                if(currentSize == batchSize){
                    channel.waitForConfirms(5000);
                    currentSize=0;
                }
            }

            if(currentSize !=0 ){
                channel.waitForConfirms(5000);
            }

            long end = System.currentTimeMillis();
            System.out.println("individually 花费： " + (end - start) + "ms");
        }
    }



    public static void publishMessagesAsynchronously() throws Exception{
        try (Connection connection = creatConnection()) {
            Channel channel = connection.createChannel();

            // 开启信道确认模式
            channel.confirmSelect();
            channel.queueDeclare(Constant.ASYNCHRONOUSLY_QUEUE, false, false, false, null);

            long start = System.currentTimeMillis();

            // 用来存储 未确认的 deliveryTag
//            SortedSet<Long> confirmSet=new TreeSet<>();   // 这种写法会报错，信息还未发送完通道关闭导致报错
            SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());
            // 添加一个监听器用来 实现异步确认

            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple){
                        // 小于等于当前序列的 deliveryTag 进行删除
                        confirmSet.headSet(deliveryTag+1).clear();
                    }else{
                        confirmSet.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.err.println("deliveryTag : "+deliveryTag +" , multiple="+multiple);
                    if(multiple){
                        System.out.println("出现错误");
                        confirmSet.headSet(deliveryTag+1).clear();
                    }else{
                        confirmSet.remove(deliveryTag);
                    }

                    // 处理完之后这里进行 发送出错的 逻辑代码的编写
                }
            });

            for (int i = 0; i < len; i++) {
                String msg = " hello I'm individually confirm " + i;
                // 获取 deliveryTag
                long nextPublishSeqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("", Constant.ASYNCHRONOUSLY_QUEUE, null, msg.getBytes());
                confirmSet.add(nextPublishSeqNo);
            }

            if(!confirmSet.isEmpty()){
                Thread.sleep(10);
            }

            long end = System.currentTimeMillis();
            System.out.println("Asynchronously 花费： " + (end - start) + "ms");
        }
    }
}


